package com.example.kafkalearn;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @author He Changjie on 2021/6/28
 */
public class AdminClientConfigOperation {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        describeTopicConfig();
//        addTopicPartitions();
    }



    /**
     * 输出主题配置
     * @throws ExecutionException
     * @throws InterruptedException
     */
    private static void describeTopicConfig() throws ExecutionException, InterruptedException {
        String brokerList = "10.51.34.61:9092";
        String topic = "harris";

        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        properties.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        AdminClient client = AdminClient.create(properties);

        ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
        DescribeConfigsResult result = client.describeConfigs(Collections.singleton(resource));
        System.out.println(result.all().get().get(resource));
        client.close();
    }
    /**
     * 分区数量只能增加，不能减少
     * @throws ExecutionException
     * @throws InterruptedException
     */
    private static void addTopicPartitions() throws ExecutionException, InterruptedException {
        String brokerList = "10.51.34.61:9092";
        String topic = "harris";

        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        properties.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        AdminClient client = AdminClient.create(properties);

        NewPartitions newPartitions = NewPartitions.increaseTo(3);
        Map<String, NewPartitions> newPartitionsMap = new HashMap<>();
        newPartitionsMap.put(topic, newPartitions);
        CreatePartitionsResult partitions = client.createPartitions(newPartitionsMap);
        partitions.all().get();

        client.close();
    }
}
